//
// Created by Alex on 17/3/10.
//

#include <sys/socket.h>
#include <list>
#include <iostream>
#include <algorithm>

#include "libev.h"
#include "uthread.h"

struct SpawnContext;
static thread_local EventLoop g_eventLoop;
static thread_local ObjectPoll<SpawnContext> g_spawnCtxPool(128);
static thread_local int g_uthreadLocalId = 0;

struct SpawnContext {
    int m_fd;
    int m_events;
    int m_currentEvent;
    uthread_t *m_uthread;
    IoEvent m_ioEv;
    TimeEvent m_timeoutEv;
};

struct uthread_cond_item_t {
    TimeEvent m_timeEv;
    AsyncEvent m_asyncEv;
    uthread_t *m_ut;
};

int cr_alloc_local_id() {
    return __sync_add_and_fetch(&g_uthreadLocalId, 1);
}

static void OnCondSignal(EventLoop *loop, Event *ev, int revent) {
    uthread_cond_item_t *it = (uthread_cond_item_t *) ev->m_data;
    if (it != nullptr) {
        uthread_t *ut = it->m_ut;
        if (ut != nullptr) {
            it->m_ut = nullptr;
            cr_resume(ut);
        }
    }
}

void cr_cond_signal(cr_cond_t *cond) {
    if (!cond->m_list.empty()) {
        uthread_cond_item_t *it = (uthread_cond_item_t *) cond->m_list[0];
        g_eventLoop.SendAsync(&(it->m_asyncEv));
    }
}

void cr_cond_broadcast(cr_cond_t *cond) {
    for (int i = 0; i < cond->m_list.size(); ++i) {
        uthread_cond_item_t *it = (uthread_cond_item_t *) cond->m_list[i];
        g_eventLoop.SendAsync(&(it->m_asyncEv));
    }
}

void cr_cond_timedwait(cr_cond_t *cond, uint64_t ms) {
    uthread_cond_item_t *it = new uthread_cond_item_t;
    it->m_timeEv.m_data = it;
    it->m_asyncEv.m_data = it;
    it->m_ut = cr_get_curr_uthread();

    if (ms > 0) {
        TimerInit(&(it->m_timeEv), OnCondSignal, ms);
        g_eventLoop.TimeStart(&(it->m_timeEv));
    }

    AsyncInit(&(it->m_asyncEv), OnCondSignal);
    g_eventLoop.StartAsync(&(it->m_asyncEv));

    cond->m_list.push_back(it);

    cr_yield();

    if (cond->m_list.size() == 1) {
        cond->m_list.pop_back();

    } else {
        for (size_t i = 0; i < cond->m_list.size(); ++i) {
            if (cond->m_list[i] == it) {
                std::swap(cond->m_list[i], cond->m_list[cond->m_list.size() - 1]);
                cond->m_list.pop_back();
            }
        }
    }

    if (ms > 0) {
        g_eventLoop.TimeStop(&(it->m_timeEv));
    }
    g_eventLoop.StopAsync(&(it->m_asyncEv));
    delete it;
}

void cr_enable_hook() {
    uthread_t *ut = cr_get_curr_uthread();
    if (ut != nullptr) {
        ut->m_enableHookSys = true;
    }
}

void cr_disable_hook() {
    uthread_t *ut = cr_get_curr_uthread();
    if (ut != nullptr) {
        ut->m_enableHookSys = false;
    }
}

void OnSpawnCtxPoll(EventLoop *loop, Event *event, int revent) {
    SpawnContext *ctx = (SpawnContext *) event->m_data;
    ctx->m_currentEvent = revent;
    cr_resume(ctx->m_uthread);
}

int cr_init(int fd) {
    fcntl(fd, F_SETFL, O_NONBLOCK);
}

int cr_accept(int fd, struct sockaddr *cliAddr, socklen_t *addrLen, uint64_t timeout) {
    int connFd = -1;
    while (1) {
        connFd = accept(fd, cliAddr, addrLen);
        if (connFd < 0) {
            if (errno == EINTR) {
                continue;
            } else if (errno == EAGAIN) {
                int event = kReadEvent;
                bool pollRet = cr_poll(fd, event, timeout);
                if (!pollRet) {
                    break;
                }
                continue;
            }
        }
        break;
    }
    return connFd;
}

void cr_sleep(uint64_t timeout_ms) {
    int event = kTimeoutEvent;
    int64_t remainTime = timeout_ms;
    if(remainTime <= 0) {
        return;
    }
    uint64_t lastAt = GetNow() + timeout_ms;

    while (remainTime > 0) {
        cr_poll(-1, event, remainTime);
        remainTime = lastAt - GetNow();

        if (timeout_ms > 0 && remainTime <= 0) //剩下时间
        {
            break;
        }
    }
}

int cr_connect(int fd, const struct sockaddr *addr, socklen_t addrLen, uint64_t timeout) {
    int ret = connect(fd, addr, addrLen);

    if (!(ret < 0 && errno == EINPROGRESS)) {
        return ret;
    }

    int event = kWriteEvent;
    int pollRet = cr_poll(fd, event, timeout);

    if (pollRet && event & kWriteEvent) {
        errno = 0;
        return 0;
    }

    int err = 0;
    socklen_t errLen = sizeof(err);
    getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errLen);
    if (err) {
        errno = err;
    } else {
        errno = ETIMEDOUT;
    }
    return ret;
}

ssize_t cr_write(int fd, const void *buf, size_t count, uint64_t timeout) {
    ssize_t left = count;
    ssize_t offset = 0;
    uint64_t remainTime = timeout;
    uint64_t lastAt = GetNow() + timeout;

    while (left > 0) {
        ssize_t ret = write(fd, (char *) buf + offset, left);
        if (ret <= 0) {
            if (errno == EINTR) {
                continue;
            } else if (errno == EAGAIN) {
                int event = kWriteEvent;
                bool pollRet = cr_poll(fd, event, remainTime);
                if (!pollRet) {
                    break;
                }

                remainTime = lastAt - GetNow();
                if (timeout > 0 && remainTime <= 0) //剩下时间
                {
                    break;
                }

                continue;
            } else {
                break;
            }
        }

        left = left - ret;
        offset = offset + ret;
    }
    return left <= 0 ? count : offset;
}

ssize_t cr_read(int fd, void *buf, size_t count, uint64_t timeout) {
    ssize_t readBytes = 0;
    uint64_t remainTime = timeout;
    uint64_t lastAt = GetNow() + timeout;
    while (1) {
        readBytes = read(fd, buf, count);
        if (readBytes >= 0) {
            break;
        } else if (errno == EINTR) {
            continue;
        } else if (errno == EAGAIN) {

            int event = kReadEvent;
            bool pollRet = cr_poll(fd, event, remainTime);
            if (!pollRet) {
                break;
            }

            remainTime = lastAt - GetNow();
            if (timeout > 0 && remainTime <= 0) //剩下时间
            {
                break;
            }

            continue;
        }
        break;
    }
    return readBytes;
}

ssize_t cr_sendto(int fd, const void *message, size_t length,
                  int flags, const struct sockaddr *dest_addr,
                  socklen_t dest_len, uint64_t timeout) {
    ssize_t ret = sendto(fd, message, length, flags, dest_addr, dest_len);
    if (ret < 0 && EAGAIN == errno) {
        int event = kWriteEvent;
        bool pollRet = cr_poll(fd, event, timeout);
        if (!pollRet) {
            return -1;
        }

        ret = sendto(fd, message, length, flags, dest_addr, dest_len);
    }
    return ret;
}

ssize_t cr_recvfrom(int fd, void *buffer, size_t length,
                    int flags, struct sockaddr *address,
                    socklen_t *address_len, uint64_t timeout) {
    ssize_t ret = recvfrom(fd, buffer, length, flags, address, address_len);
    if (ret < 0 && EAGAIN == errno) {
        int event = kReadEvent;
        bool pollRet = cr_poll(fd, event, timeout);
        if (!pollRet) {
            return -1;
        }

        ret = recvfrom(fd, buffer, length, flags, address, address_len);
    }
    return ret;

}

bool cr_poll(int fd, int &events, uint64_t timeout) {
    bool validIoEv = fd > 0 && ((events & kWriteEvent) || (events & kReadEvent));
    bool validTimeoutEv = (timeout > 0);

    uthread_t *ut = cr_get_curr_uthread();
    if (ut == nullptr || (!validIoEv && !validTimeoutEv)) {
        return -1;
    }

    SpawnContext *ctx = g_spawnCtxPool.AllocObj();
    ctx->m_fd = fd;
    ctx->m_events = events;
    ctx->m_currentEvent = kNone;
    ctx->m_uthread = ut;

    ctx->m_ioEv.m_data = ctx;
    ctx->m_timeoutEv.m_data = ctx;

    if (validIoEv) {
        IoInit(&ctx->m_ioEv, OnSpawnCtxPoll, ctx->m_fd, ctx->m_events);
        g_eventLoop.IoStart(&ctx->m_ioEv);
    }

    if (validTimeoutEv) {
        TimerInit(&ctx->m_timeoutEv, OnSpawnCtxPoll, timeout);
        g_eventLoop.TimeStart(&ctx->m_timeoutEv);
    }


    cr_yield();


    //io事件
    bool isIoEv = validIoEv;
    bool isTimeEv = !validIoEv && validTimeoutEv;

    bool ret = false;
    if (isIoEv) {
        if ((ctx->m_currentEvent & kReadEvent) || (ctx->m_currentEvent & kWriteEvent)) {
            ret = true;
        }
    } else if (isTimeEv && (ctx->m_currentEvent & kTimeoutEvent)) {
        ret = true;
    }

    if (validIoEv) {
        g_eventLoop.IoStop(&ctx->m_ioEv);
    }
    if (validTimeoutEv) {
        g_eventLoop.TimeStop(&ctx->m_timeoutEv);
    }

    events = ctx->m_currentEvent;
    g_spawnCtxPool.FreeObj(ctx);
    return ret;
}

void cr_scheduler()
{
    g_eventLoop.Dispath();
}

/**
 * 读数据读入, 直到可读
 * @param fd
 * @param iov
 * @param iovcnt
 * @return
 */
ssize_t cr_readv(int fd, const struct iovec *iov, int iovcnt) {
    ssize_t ret = readv(fd, iov, iovcnt);
    if (ret <= 0 && EAGAIN == errno) {
        int event = kReadEvent;

        bool pollRet = cr_poll(fd, event, 0);
        if (!pollRet) {
            return -1;
        }

        ret = readv(fd, iov, iovcnt);
    }
    return ret;
}

/**
 * 把数据写出, 直到可写
 * @param fd
 * @param iov
 * @param iovcnt
 * @return
 */
ssize_t cr_writev(int fd, const struct iovec *iov, size_t iovcnt) {
    ssize_t ret = writev(fd, iov, iovcnt);
    if (ret < 0 && EAGAIN == errno) {
        int event = kWriteEvent;

        bool pollRet = cr_poll(fd, event, 0);
        if (!pollRet) {
            return -1;
        }

        ret = writev(fd, iov, iovcnt);
    }
    return ret;
}

/**
 * 把所有数据写出, 直到出错或写入结束
 * @param fd
 * @param iov
 * @param iovcnt
 * @return
 */
ssize_t cr_writev_all(int fd, struct iovec *iov, size_t iovcnt) {
    size_t totalLen = 0;
    size_t *needOut = new size_t[iovcnt];

    for (size_t i = 0; i < iovcnt; ++i) {
        totalLen += iov->iov_len;
        needOut[i] = totalLen;
    }

    ssize_t writeRet = 0;
    ssize_t writeLen = 0;
    size_t startPos = 0;

    do {
        size_t writeCnt = iovcnt - startPos < 1 ? iovcnt - startPos : 1;
        writeRet = cr_writev(fd, iov + startPos, writeCnt);

        if (writeRet <= 0) {
            break;
        }

        writeLen += writeRet;

        if (writeLen >= totalLen) {
            break;
        }

        size_t pos =
                std::upper_bound(needOut + startPos,
                                 needOut + iovcnt, writeLen) - needOut;
        startPos = pos;

        size_t len = writeLen + iov[pos].iov_len;
        if (len > needOut[pos]) {
            size_t l = len - needOut[pos]; //已经写出多少
            iov[pos].iov_base = (char *) iov[pos].iov_base + l;
            iov[pos].iov_len -= l;
        }

    } while (writeLen < totalLen);

    delete[] needOut;

    return writeLen >= totalLen ? writeLen : -1;
}

